package com.lkd.client.emq;

import com.lkd.client.config.EmqConfig;
import com.lkd.client.utils.SslUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class EmqClient {
    @Autowired
    private EmqConfig emqConfig;

    private MqttClient mqttClient;

    /**
     * 连接mqtt borker
     */
    public void connect(MqttCallback callback) {
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            mqttClient = new MqttClient(emqConfig.getMqttServerUrl(), emqConfig.getClientId(), persistence);
        } catch (MqttException e) {
            log.error("mqtt creat error", e);
        }
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setAutomaticReconnect(false);
        connOpts.setUserName("test");
        connOpts.setPassword(emqConfig.getMqttPassword().toCharArray());
        connOpts.setCleanSession(false);
        connOpts.setAutomaticReconnect(true);
//        try {
//            // 设置SSL加密链接
//            connOpts.setSocketFactory(SslUtil.getSocketFactory("C:\\Users\\LIU\\Desktop\\zs\\ca.crt", "C:\\Users\\LIU\\Desktop\\zs\\client.crt", "C:\\Users\\LIU\\Desktop\\zs\\client.key", "123456"));
//
//        } catch (Exception e) {
//            log.info("证书读取失败:{}", e);
//        }
        try {
            mqttClient.setCallback(callback);
            mqttClient.connect(connOpts);
        } catch (MqttException e) {
            log.error("mqtt creat error", e);
        }
    }


    /**
     * 订阅及回调方法
     *
     * @param topicFilter
     * @throws MqttException
     */
    public void subscribe(String topicFilter, MqttCallback callBack) throws MqttException {
        log.info("subscribe----------- {}", topicFilter);
        mqttClient.subscribe(topicFilter, 1);
        mqttClient.setCallback(callBack);
    }

    /**
     * 发布消息
     *
     * @param msgType 协议类型
     * @param msg     消息体
     */
    @Async//异步方法的注解
    public void publish(String msgType, String msg) {
        try {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(1);
            mqttMessage.setRetained(false);
            mqttMessage.setPayload(msg.getBytes());//设置消息体
            log.info("发布消息的主题是 :{}", emqConfig.getPublisTopicPrefix()+msgType);
            log.info("发布消息体是2 :{}", msg);
            MqttTopic mTopic = mqttClient.getTopic(emqConfig.getPublisTopicPrefix()+msgType);
            if (null == mTopic) {
                log.error("topic is not exist");
            }
            MqttDeliveryToken token;
            try {
                token = mTopic.publish(mqttMessage);
                token.waitForCompletion();
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            }
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("mqtt publish msg error", e);
        }
    }


    /**
     * 发布消息
     *
     * @param msg     消息体
     */
    @Async//异步方法的注解
    public void publish(String msg) {
        try {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(1);
            mqttMessage.setRetained(false);
            mqttMessage.setPayload(msg.getBytes());//设置消息体
            log.info("发布消息的主题是 :{}", emqConfig.getFirstTopic());
            log.info("发布消息体是 msg:{}", msg);
            MqttTopic mTopic = mqttClient.getTopic(emqConfig.getFirstTopic());
            if (null == mTopic) {
                log.error("topic is not exist");
            }
            MqttDeliveryToken token;
            try {
                token = mTopic.publish(mqttMessage);
                token.waitForCompletion();
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            }
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("mqtt publish msg error", e);
        }
    }

}
